package com.sunday.common.mq.rocket.brave.study.base.e6_Transaction_Message;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 2、实现事务的监听接口
 * <p>
 * 本示例主要说明的是 当 ROLLBACK_MESSAGE 时候直接丢弃， 而 COMMIT_MESSAGE 会立即进行消费，
 * 优先执行 executeLocalTransaction 其中返回 UNKNOW 后，将进行 checkLocalTransaction 的定期验证功能，直到获得明确结果，结果如果是 COMMIT_MESSAGE，此时 Consumer才会进行消费，
 * 可以与业务进紧密配合，获取最终是否进行操作。
 *
 * <p>
 * RocketMQ的TransactionListener是用于处理分布式事务消息的接口。在使用事务消息时，可以通过实现TransactionListener接口来定义事务消息的执行逻辑。
 * 该接口包含两个方法：
 * <p>
 * executeLocalTransaction：此方法用于执行本地事务。当发送事务消息时，RocketMQ会调用该方法来执行与事务相关的本地业务操作，
 * 例如数据库的更新或其他操作。在该方法中，您可以执行必要的业务逻辑，并返回一个事务状态（COMMIT、ROLLBACK或UNKNOWN）给RocketMQ。
 * <p>
 * checkLocalTransaction：此方法用于检查本地事务状态。RocketMQ在一段时间后会调用该方法来检查本地事务的状态。在该方法中，
 * 您需要根据消息的状态来判断事务是否已经提交或回滚，并返回相应的事务状态给RocketMQ。
 * <p>
 * 通过实现TransactionListener接口并配置给Producer，在发送事务消息时，RocketMQ会根据您的业务逻辑执行本地事务，
 * 并根据checkLocalTransaction方法的返回值来确认最终的事务状态。
 * 这样可以确保事务消息的可靠性和一致性，同时提供了较高的扩展性和灵活性。
 * <p>
 * 使用RocketMQ的TransactionListener对性能会有一定的影响，因为它涉及到执行本地事务和检查本地事务状态的过程。
 * 在发送事务消息时，RocketMQ会调用executeLocalTransaction方法来执行本地事务。该方法内部执行的业务逻辑可能需要一定的时间和资源，具体取决于实际的本地事务操作。如果本地事务操作很复杂或耗时较长，可能会对性能产生一定的压力。
 * 另外，RocketMQ还会周期性地调用checkLocalTransaction方法来检查本地事务的状态。如果消息长时间未提交或回滚，可能会导致频繁地调用checkLocalTransaction方法，增加了一定的开销和性能压力。
 * 为了减轻TransactionListener对性能的压力，可以考虑以下几点：
 * 优化本地事务逻辑：确保本地事务操作尽量快速和高效，避免执行过长或复杂的操作，减少对性能的影响。
 * 合理设置消息检查间隔：通过合理设置RocketMQ的参数，如transactionCheckInterval，来平衡事务消息的可靠性和性能开销。较长的检查间隔可以减少checkLocalTransaction方法的调用次数，从而减轻对性能的影响。
 * 考虑异步处理：如果本地事务操作非常耗时，可以考虑将其异步化，例如使用异步线程池或消息队列来处理事务逻辑。这样可以减少主线程的阻塞时间，提高系统的并发性能。
 * 总之，TransactionListener的性能压力取决于具体的业务逻辑和消息处理需求。合理优化本地事务操作、调整检查间隔，并结合实际场景做出相应的性能调整，可以更好地平衡可靠性和性能。
 * <p>
 * 6.2 事务消息使用上的限制
 * 事务消息不支持延时消息和批量消息。
 * 为了避免单个消息被检查太多次而导致半队列消息累积，我们默认将单个消息的检查次数限制为 15 次，但是用户可以通过 Broker 配置文件的 transactionCheckMax参数来修改此限制。如果已经检查某条消息超过 N 次的话（ N = transactionCheckMax ） 则 Broker 将丢弃此消息，并在默认情况下同时打印错误日志。用户可以通过重写 AbstractTransactionalMessageCheckListener 类来修改这个行为。
 * 事务消息将在 Broker 配置文件中的参数 transactionTimeout 这样的特定时间长度之后被检查。当发送事务消息时，用户还可以通过设置用户属性 CHECK_IMMUNITY_TIME_IN_SECONDS 来改变这个限制，该参数优先于 transactionTimeout 参数。
 * 事务性消息可能不止一次被检查或消费。
 * 提交给用户的目标主题消息可能会失败，目前这依日志的记录而定。它的高可用性通过 RocketMQ 本身的高可用性机制来保证，如果希望确保事务消息不丢失、并且事务完整性得到保证，建议使用同步的双重写入机制。
 * 事务消息的生产者 ID 不能与其他类型消息的生产者 ID 共享。与其他类型的消息不同，事务消息允许反向查询、MQ服务器能通过它们的生产者 ID 查询到消费者。
 */
@Slf4j
public class TransactionListenerImpl implements TransactionListener {

    private Random random = new Random();
    private AtomicInteger transactionIndex = new AtomicInteger(0);
    private ConcurrentHashMap<String, Integer> localTrans = new ConcurrentHashMap<>();

    /**
     * 当发送事务准备(半)消息成功时，将调用此方法执行本地事务。
     * 状态机对应
     * 0 -> UNKNOW              中间状态，它代表需要检查消息队列来确定状态。
     * 1 -> COMMIT_MESSAGE      提交事务，它允许消费者消费此消息。
     * 2 -> ROLLBACK_MESSAGE    回滚事务，它代表该消息将被删除，不允许被消费。
     *
     * @param msg 一半(准备)消息
     * @param arg 自定义业务参数
     * @return 事务状态
     */
    @Override
    public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        // 方法来执行本地事务。它返回前一节中提到的三个事务状态之一。
        int value = transactionIndex.getAndIncrement();
        int status = value % 3;
        log.info("executeLocalTransaction TransactionId : {} , Status : {}", msg.getTransactionId(), status);
        localTrans.put(msg.getTransactionId(), status);
        return LocalTransactionState.UNKNOW;
    }

    /**
     * 没有回应时准备(半)留言。Broker将发送check消息来检查事务状态，并以此
     * 方法将被调用以获取本地事务状态。
     * TransactionStatus.CommitTransaction: 提交事务，它允许消费者消费此消息。
     * TransactionStatus.RollbackTransaction: 回滚事务，它代表该消息将被删除，不允许被消费。
     * TransactionStatus.Unknown: 中间状态，它代表需要检查消息队列来确定状态。
     *
     * @param msg 检查消息
     * @return 事务状态
     */
    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt msg) {
        // 方法用于检查本地事务状态，并回应消息队列的检查请求。它也是返回前一节中提到的三个事务状态之一。
        Integer status = localTrans.get(msg.getTransactionId());
        if (null != status) {
            switch (status) {
                case 0:
                    log.info("checkLocalTransaction TransactionId : {} , old Status : {}", msg.getTransactionId(), LocalTransactionState.UNKNOW);
                    if (random.nextBoolean()) {
                        return LocalTransactionState.UNKNOW;
                    } else {
                        log.info("checkLocalTransaction TransactionId : {} , new Status : {}", msg.getTransactionId(), LocalTransactionState.COMMIT_MESSAGE);
                        return LocalTransactionState.COMMIT_MESSAGE;
                    }
                case 1:
                    log.info("checkLocalTransaction TransactionId : {} , Status : {}", msg.getTransactionId(), LocalTransactionState.COMMIT_MESSAGE);
                    return LocalTransactionState.COMMIT_MESSAGE;
                case 2:
                    log.info("checkLocalTransaction TransactionId : {} , Status : {}", msg.getTransactionId(), LocalTransactionState.ROLLBACK_MESSAGE);
                    return LocalTransactionState.ROLLBACK_MESSAGE;
            }
        }
        return LocalTransactionState.COMMIT_MESSAGE;
    }

}